package com.nulldev.util.web.HttpServer3.extensions.sse;

import java.io.Closeable;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.util.Timer;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicBoolean;

import com.nulldev.util.VariableAPI.MathUtil;
import com.nulldev.util.VariableAPI.RandomUtil;
import com.nulldev.util.concurrency.threadIt.v4.locking.ResettableCountDownLatch;
import com.nulldev.util.io.IOUtils;
import com.nulldev.util.web.HttpServer3.ReadRequest;
import com.nulldev.util.web.HttpServer3.extensions.sse.event.SSEvent;
import com.nulldev.util.web.HttpServer3.extensions.sse.event.io.SSEventStreamWriter;
import com.nulldev.util.web.HttpServer3.extensions.sse.event.io.SSEventStreamWriterImpl;
import com.nulldev.util.web.HttpServer3.util.HTTPPrintWriter;

public abstract class SSESession implements Runnable, Closeable {

	private final Socket socket;
	private final InputStream is;
	private final OutputStream os;
	private final HTTPPrintWriter writer;
	private final UUID peerID;
	private final SSEventStreamWriter sWriter;
	private final AtomicBoolean isOpen = new AtomicBoolean(true);
	private String lastEventID;
	private final Timer heartbeatTimer;
	private final ResettableCountDownLatch close = new ResettableCountDownLatch(1);

	private final ReadRequest rr;
	private final String url;

	public SSESession(final Socket socket, final InputStream is, final OutputStream os, final HTTPPrintWriter pw, final ReadRequest rr, final String url) {
		this.socket = socket;
		this.is = is;
		this.os = os;
		this.peerID = RandomUtil.quickUUID();
		this.writer = pw;
		this.sWriter = new SSEventStreamWriterImpl(pw);
		this.lastEventID = rr.headers().get("Last-Event-ID");
		this.heartbeatTimer = new Timer();
		this.rr = rr;
		this.url = url;
		try {
			this.socket.setTrafficClass(0x10); /* IPTOS_LOWDELAY */
		} catch (Exception e) {
		}
	}

	public Socket getSocket() {
		return this.socket;
	}

	public InputStream getInputStream() {
		return this.is;
	}

	public OutputStream getOutputStream() {
		return this.os;
	}

	public String getLastEventID() {
		return this.lastEventID;
	}

	public UUID getUUID() {
		return this.peerID;
	}

	@Override
	public void run() {
		try {
			this.onConnected(this.rr, this.url);
			this.close.await();
		} catch (InterruptedException e) {
			try {
				this.close();
				this.close.countDown();
			} catch (Exception ex) {
			}
		}
	}

	@Override
	public void close() throws IOException {
		IOUtils.closeQuietly(this.sWriter);
		IOUtils.closeAllQuietly(this.socket, this.is, this.os, this.writer);
		this.close.countDown();
	}

	protected Heartbeat setHeartbeat(SSESession es, long heartbeatMs) {
		Heartbeat hb = new Heartbeat(es);
		this.heartbeatTimer.scheduleAtFixedRate(hb, heartbeatMs, heartbeatMs);
		return hb;
	}

	public void sendEvent(final SSEvent event) throws IOException {
		if (!this.isOpen.get())
			throw new IllegalStateException("Connection is closed");
		this.sWriter.write(event);
		this.writer.flush();
	}

	public void sendComment(final String comment) throws IOException {
		if (!this.isOpen.get())
			throw new IllegalStateException("Connection is closed");
		this.writer.print(": " + comment + "\n");
		this.writer.print("\n");
		this.writer.flush();
	}

	public boolean isOpen() {
		return this.isOpen.get();
	}

	@Override
	public String toString() {
		return "SSESession[hash=" + MathUtil.toHex(super.hashCode()) + ",uuid=" + this.peerID + ",open=" + this.isOpen.get() + "]";
	}

	public abstract void onConnected(final ReadRequest rr, final String url);

}
